
#include <fcntl.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
#include <algorithm>
#include <cstdlib>
#include <iostream>
#include <string>
#include <thread>
#include <unordered_map>

#include "spdlog/sinks/spinlock_limit_time_sink.h"

#if __GNUC__ >= 3
#define spd_likely(x) (__builtin_expect((x), 1))
#define spd_unlikely(x) (__builtin_expect((x), 0))
#else
#define spd_likely(x) (x)
#define spd_unlikely(x) (x)
#endif

namespace spdlog {
namespace sinks {

constexpr size_t kSpinLockTryMaxTime = 10;
constexpr size_t kInitialBufCapcity = 1024 << 14;
constexpr size_t kSpdlogWrapperBufLen =
    512 - sizeof(details::log_msg) - sizeof(std::atomic<std::thread::id>);
constexpr size_t kPageLen = 4096;

namespace ThreadLocalLogDeal {
// there is too many threads
constexpr size_t kThreadStringBufferMaxSize = 64;
constexpr size_t kAllStringBufferSize = 4096 << 2;

struct AlignBuf {
  AlignBuf(size_t size) {
    auto res = posix_memalign((void**)&data, 512, size);
    if (res == 0) {
      this->size = size;
    }
  }

  ~AlignBuf() {
    if (data != nullptr) free(data);
  }

  void swap(AlignBuf& buf) {
    std::swap(buf.data, data);
    std::swap(buf.size, size);
    std::swap(buf.now_i, now_i);
  }

  size_t PushData(memory_buf_t& buf) {
    return PushRawChar(buf.data(), buf.size());
  }

  size_t PushRawChar(char* src, size_t buf_len) {
    if (buf_len + now_i > size) return 0;

    memcpy(data + now_i, src, buf_len);
    now_i += buf_len;
    return buf_len;
  }

  size_t Size() { return now_i; }

  size_t Capcity() { return size; }

  void Clear() { now_i = 0; }

  unsigned char* data{nullptr};
  size_t size{0};
  size_t now_i{0};
};
struct SpdlogMsgWrapper {
  details::log_msg spd_msg;

  char char_buf[kSpdlogWrapperBufLen];

  SpdlogMsgWrapper() {
    spd_msg.payload = string_view_t(char_buf, 0);
    char_buf[kSpdlogWrapperBufLen - 1] = '\0';
  }

  SpdlogMsgWrapper(const details::log_msg& spd_msg) { ReplaceSpdMsg(spd_msg); }

  SpdlogMsgWrapper(const SpdlogMsgWrapper& other) {
    if (&other != this) {
      spd_msg = other.spd_msg;
      memcpy(char_buf, other.char_buf, kSpdlogWrapperBufLen);
    }
  }

  void ReplaceSpdMsg(const details::log_msg& spd_msg) {
    this->spd_msg = spd_msg;
    size_t need_copy =
        std::min(spd_msg.payload.size(), kSpdlogWrapperBufLen - 2);
    memcpy((void*)(char_buf), spd_msg.payload.data(), need_copy);
    this->spd_msg.payload = string_view_t(char_buf, need_copy);
    char_buf[need_copy] = '\0';
  }
};

// todo : can we make it align the cache line?
struct ThreadSpdlogMsgBuf {
  SpdlogMsgWrapper spdlog_buf;
  std::atomic<std::thread::id> thread_id;

  ThreadSpdlogMsgBuf() : thread_id(std::thread::id()) {}
};

static ThreadSpdlogMsgBuf s_str_buf[kAllStringBufferSize];
thread_local ThreadSpdlogMsgBuf* thread_str_buf_[kThreadStringBufferMaxSize]{
    nullptr};
thread_local size_t thread_str_buf_i_{kThreadStringBufferMaxSize + 1};

ThreadSpdlogMsgBuf* AllocThreadBuf() {
  for (size_t i = 0; i < kAllStringBufferSize; ++i) {
    auto now_id = s_str_buf[i].thread_id.load(std::memory_order_acquire);
    if (now_id == std::thread::id() &&
        s_str_buf[i].thread_id.compare_exchange_strong(
            now_id, std::this_thread::get_id(), std::memory_order_release,
            std::memory_order_relaxed)) {
      return &s_str_buf[i];
    }
  }
  return nullptr;
}

bool PushData(const details::log_msg& spd_msg) {
  // to do : intialize when thread create rather than judge every time
  if (spd_unlikely(thread_str_buf_i_ > kThreadStringBufferMaxSize)) {
    for (size_t i = 0; i < kThreadStringBufferMaxSize; ++i) {
      auto buf = AllocThreadBuf();
      thread_str_buf_[i] = buf;
    }
    thread_str_buf_i_ = 0;
  }
  if (spd_unlikely(thread_str_buf_i_ == kThreadStringBufferMaxSize))
    return false;

  auto buf = thread_str_buf_[thread_str_buf_i_];
  bool ans = true;
  if (buf != nullptr) {
    buf->spdlog_buf.ReplaceSpdMsg(spd_msg);
  } else {
    ans = false;
  }
  thread_str_buf_i_++;
  return ans;
}

bool PopAllData(ThreadSpdlogMsgBuf*& start, size_t& len) {
  if (thread_str_buf_i_ > 0 &&
      thread_str_buf_i_ <= kThreadStringBufferMaxSize) {
    start = thread_str_buf_[0];
    len = thread_str_buf_i_;
    thread_str_buf_i_ = 0;
    return true;
  } else {
    return false;
  }
}

void BackThreadBuf() {
  if (thread_str_buf_i_ < kThreadStringBufferMaxSize)
    std::for_each(
        thread_str_buf_, thread_str_buf_ + kThreadStringBufferMaxSize - 1,
        [](ThreadSpdlogMsgBuf* buf) {
          buf->thread_id.store(std::thread::id(), std::memory_order_release);
        });
}

}  // namespace ThreadLocalLogDeal

void SpinLockLimitTimeSink::sink_it_(const details::log_msg& spd_msg) {
  size_t try_time = 0;
  while (flag_.test_and_set(std::memory_order_acquire)) {
    if (++try_time > kSpinLockTryMaxTime) {
      ThreadLocalLogDeal::PushData(spd_msg);
      return;
    }
  }

  ThreadLocalLogDeal::ThreadSpdlogMsgBuf* in_msg;
  size_t len = 0;
  if (spd_unlikely(ThreadLocalLogDeal::PopAllData(in_msg, len))) {
    for (size_t i = 0; i < len; ++i) {
      auto msg = in_msg + i;
      if (msg != nullptr) {
        memory_buf_t formatted;
        formatter_->format(msg->spdlog_buf.spd_msg, formatted);
        active_buf_->PushData(formatted);
      }
    }
  }

  if (spd_unlikely(active_buf_->Size() >= kInitialBufCapcity)) {
    ThreadLocalLogDeal::PushData(spd_msg);
  } else {
    memory_buf_t formatted;
    // formating spends too much time that formatting cost 2us while all other
    // operation spending below 1us so NL_INFO << maybe faster that LOG_INFO()
    formatter_->format(spd_msg, formatted);
    active_buf_->PushData(formatted);
  }
  flag_.clear(std::memory_order_release);
}

SPDLOG_INLINE SpinLockLimitTimeSink::SpinLockLimitTimeSink(
    filename_t base_filename, std::size_t max_size, std::size_t max_files,
    RotatingFunction func, bool rotate_on_open,
    const file_event_handlers& event_handlers)
    : base_filename_(std::move(base_filename)),
      max_size_(max_size),
      max_files_(max_files),
      file_helper_(),
      file_fd_(-1),
      rotate_func_(std::move(func)),
      rotate_on_open_(rotate_on_open),
      formatter_{details::make_unique<spdlog::pattern_formatter>()},
      last_write_is_ok_(true) {
  if (max_size == 0) {
    throw_spdlog_ex("rotating sink constructor: max_size arg cannot be zero");
  }

  if (max_files > 200000) {
    throw_spdlog_ex(
        "rotating sink constructor: max_files arg cannot exceed 200000");
  }

  auto file_name = rotate_func_();
  for (size_t i = 0; i < 3; ++i) {
    file_fd_ = open(file_name.c_str(), O_WRONLY | O_CREAT, 0755);
    if (file_fd_ > 0) break;
  }

  if (file_fd_ < 0) {
    throw_spdlog_ex("open file directly error");
  }

  int ret = unlink(base_filename_.c_str());
  if (ret != 0) {
    std::cerr << "delete link error " << base_filename_ << std::endl;
  }
  ret = symlink(file_name.c_str(), base_filename_.c_str());
  if (ret != 0) {
    std::cerr << "create link error " << base_filename_ << " -> " << file_name
              << std::endl;
  }

  active_buf_ =
      std::make_unique<ThreadLocalLogDeal::AlignBuf>(kInitialBufCapcity);
  flushing_buf_ =
      std::make_unique<ThreadLocalLogDeal::AlignBuf>(kInitialBufCapcity);
  StartNewThread();
}

SpinLockLimitTimeSink::~SpinLockLimitTimeSink() {
  file_helper_.close();
  need_end_ = true;
  write_thread_->join();
}

void SpinLockLimitTimeSink::Rotate() {
  using details::os::filename_to_str;
  using details::os::path_exists;

  if (file_fd_ != -1) close(file_fd_);

  auto file_name = rotate_func_();
  file_fd_ = open(file_name.c_str(), O_WRONLY | O_CREAT, 0755);

  unlink(base_filename_.c_str());
  symlink(file_name.c_str(), base_filename_.c_str());
}

void SpinLockLimitTimeSink::ThreadFunc() {
  std::this_thread::sleep_for(std::chrono::seconds(1));
  while (!need_end_) {
    size_t try_time = 0;
    while (flag_.test_and_set(std::memory_order_acquire)) {
      if (++try_time % 64 == 0) std::this_thread::yield();
    }
    // actually, even one align buf is ok, direct IO operate whole page rather
    // than random address
    if (last_write_is_ok_ && BufSizeIsOk(active_buf_))
      active_buf_.swap(flushing_buf_);
    flag_.clear(std::memory_order_release);
    WriteBuffer(flushing_buf_);
    if (flushing_buf_->Size() < kInitialBufCapcity >> 4) {
      // please estimate the speed of creating log to determine the time of
      // sleep
      std::this_thread::sleep_for(std::chrono::milliseconds(200));
    } else {
      std::this_thread::sleep_for(std::chrono::microseconds(50));
    }
  }
}

bool SpinLockLimitTimeSink::BufSizeIsOk(
    const std::unique_ptr<ThreadLocalLogDeal::AlignBuf>& buf) {
  auto len = buf->Size();
  auto shift_len = len - len % kPageLen;
  auto discard_len = len - shift_len;

  return (len > 2 << 12);
}

void SpinLockLimitTimeSink::StartNewThread() {
  write_thread_ = std::make_unique<std::thread>([this]() { ThreadFunc(); });
}

void SpinLockLimitTimeSink::WriteBuffer(
    const std::unique_ptr<ThreadLocalLogDeal::AlignBuf>& msgs) {
  static size_t write_msg_num{0};
  char page_data[kPageLen]{'\0'};
  auto len = msgs->Size();
  auto offset_len = len % kPageLen;
  auto shift_len = len - offset_len;

  if (len > 2 << 12) {
    auto write_ans = write(file_fd_, msgs->data, len);
    if (write_ans > 0) {
      msgs->Clear();
      write_msg_num += len;
      last_write_is_ok_ = true;
    } else {
      last_write_is_ok_ = false;
    }
  }

  if (write_msg_num > max_size_) {
    Rotate();
    write_msg_num = 0;
  }
}
}  // namespace sinks

}  // namespace spdlog
